package com.fanxl.flink.datastream.cep;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @description: https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/libs/cep/
 * @author: fanxiaole
 * @date: 2022/3/29 9:46
 */
public class CepApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        test1(env);
        env.execute("CepApp");
    }

    /**
     * 连续出现2次登录失败
     * @param env
     */
    private static void test1(StreamExecutionEnvironment env) {
        env.setParallelism(1);

        KeyedStream<Access, String> stream = env.fromElements(
                "001, 78.89.90.2, success,1622689918",
                "002, 110.111.112.113, failure,1622689952",
                "002, 110.111.112.113, failure,1622689953",
                "002, 110.111.112.113, failure,1622689954",
                "002, 193.114.45.13, success,1622689956",
                "002, 137.49.24.26, failure,1622689955"
        ).map(new MapFunction<String, Access>() {
            @Override
            public Access map(String value) throws Exception {
                String[] array = value.split(",");
                Access access = new Access();
                access.userId = array[0];
                access.ip = array[1].trim();
                access.result = array[2].trim();
                access.time = Long.parseLong(array[3].trim());
                return access;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<Access>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((access, timestamp) -> access.time * 1000))
                .keyBy(x -> x.userId);

        // 1) 定义模式
        Pattern<Access, Access> pattern = Pattern.<Access>begin("start")
                .where(new SimpleCondition<Access>() {
                    @Override
                    public boolean filter(Access value) throws Exception {
                        return value.result.equals("failure");
                    }
                })
                .next("next")
                .where(new SimpleCondition<Access>() {
                    @Override
                    public boolean filter(Access value) throws Exception {
                        return value.result.equals("failure");
                    }
                })
                // 两条数据之间的间隔 超过就不符合
                .within(Time.seconds(3));

        // 把规则作用到数据流
        PatternStream<Access> patternStream = CEP.pattern(stream, pattern);

        // 2) 匹配结果
        patternStream.select(new PatternSelectFunction<Access, AccessMsg>() {
            @Override
            public AccessMsg select(Map<String, List<Access>> map) throws Exception {
                Access first = map.get("start").get(0);
                Access last = map.get("next").get(0);
                AccessMsg accessMsg = new AccessMsg();
                accessMsg.first = first.time;
                accessMsg.userId = first.userId;
                accessMsg.second = last.time;
                accessMsg.msg = "连续出现登录失败";
                return accessMsg;
            }
        }).print();


    }
}

class AccessMsg {
    public String userId;

    public Long first;

    public Long second;

    public String msg;

    @Override
    public String toString() {
        return "AccessMsg{" +
                "userId='" + userId + '\'' +
                ", first=" + first +
                ", second=" + second +
                ", msg='" + msg + '\'' +
                '}';
    }
}

class Access {
    public String userId;

    public String ip;

    public String result;

    public Long time;
}
